/*
 * 
 * 
 * 
 * 
 * 
 * the License.  
 *
 *    
 *
 * 
 * 
 * 
 * 
 * 
 */

package drds.server.memory.unsafe.utils.sort;

import java.io.BufferedInputStream;
import java.io.Closeable;
import java.io.DataInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;

import com.google.common.io.ByteStreams;
import com.google.common.io.Closeables;

import drds.server.memory.unsafe.Platform;
import drds.server.memory.unsafe.storage.ConnectionId;
import drds.server.memory.unsafe.storage.SerializerManager;

/**
 * Reads spill files written by {@link UnsafeSorterSpillWriter} (see that class
 * for a description of the file format).
 */
public final class UnsafeSorterSpillReader extends UnsafeSorterIterator implements Closeable {

	private InputStream in;
	private DataInputStream din;

	// Variables that change with every record read:
	private int recordLength;
	private long keyPrefix;
	private int numRecords;
	private int numRecordsRemaining;

	private byte[] arr = new byte[1024 * 1024];
	private Object baseObject = arr;
	private final long baseOffset = Platform.BYTE_ARRAY_OFFSET;

	public UnsafeSorterSpillReader(SerializerManager serializerManager, File file, ConnectionId blockId) throws IOException {
		assert (file.length() > 0);
		final BufferedInputStream bs = new BufferedInputStream(new FileInputStream(file));
		try {
			this.in = serializerManager.wrapForCompression(blockId, bs);
			this.din = new DataInputStream(this.in);
			numRecords = numRecordsRemaining = din.readInt();
		} catch (IOException e) {
			Closeables.close(bs, /* swallowIOException = */true);
			throw e;
		}
	}

	@Override
	public int getNumRecords() {
		return numRecords;
	}

	@Override
	public boolean hasNext() {
		return (numRecordsRemaining > 0);
	}

	@Override
	public void loadNext() throws IOException {
		recordLength = din.readInt();
		keyPrefix = din.readLong();
		if (recordLength > arr.length) {
			arr = new byte[recordLength];
			baseObject = arr;
		}
		ByteStreams.readFully(in, arr, 0, recordLength);
		numRecordsRemaining--;
		if (numRecordsRemaining == 0) {
			close();
		}
	}

	@Override
	public Object getBaseObject() {
		return baseObject;
	}

	@Override
	public long getBaseOffset() {
		return baseOffset;
	}

	@Override
	public int getRecordLength() {
		return recordLength;
	}

	@Override
	public long getKeyPrefix() {
		return keyPrefix;
	}

	@Override
	public void close() throws IOException {
		if (in != null) {
			try {
				in.close();
			} finally {
				in = null;
				din = null;
			}
		}
	}
}
